package com.shujia.ads;

import com.alibaba.alink.common.MLEnvironmentFactory;
import com.alibaba.alink.pipeline.PipelineModel;
import com.shujia.function.DateFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import com.alibaba.alink.operator.stream.StreamOperator;

public class Demo5SentIndex1 {
    public static void main(String[] args) throws Exception {


        StreamTableEnvironment env = MLEnvironmentFactory.getDefault().getStreamTableEnvironment();

        /*
         * 老版本建表方式
         *
         */
        env.executeSql("" +
                "CREATE TABLE dwd_kafka_comment_msk (\n" +
                "  comment_id String,\n" +
                "  comment_time String,\n" +
                "  like_count bigint,\n" +
                "  comment_text String,\n" +
                "  user_id String,\n" +
                "  weibo_id String\n" +
                ") WITH (\n" +
                "  'connector.type' = 'kafka',       \n" +
                "  'connector.version' = 'universal', \n" +
                "  'connector.topic' = 'dwd_kafka_comment_msk', \n" +
                "  'connector.properties.zookeeper.connect' = 'master:2181,node1:2181,node2:2181', \n" +
                "  'connector.properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',\n" +
                "  'connector.properties.group.id' = 'ADasASA',\n" +
                "  'connector.startup-mode' = 'earliest-offset',\n" +
                "  'format.type' = 'json' \n" +
                ")");

        // register function
        env.createTemporarySystemFunction("dateFunction", DateFunction.class);


        Table table = env.sqlQuery("select comment_text as text ,dateFunction(comment_time)  as d from dwd_kafka_comment_msk");


        /*
         * 加载模型
         *
         */

        PipelineModel model = PipelineModel.load("data/model");

        //情感打标
        Table transform = model.transform(table);


        env.createTemporaryView("temp", transform);


        /*
         * 将结果保存到mysql
         *
         */
        env.executeSql("" +
                "CREATE TABLE ads_mysql_sent_index (\n" +
                "  sent double,\n" +
                "  d String,\n" +
                "  c bigint\n" +
                ") WITH (\n" +
                "  'connector.type' = 'jdbc',\n" +
                "  'connector.url' = 'jdbc:mysql://master:3306/sent',\n" +
                "  'connector.table' = 'ads_mysql_sent_index', \n" +
                "  'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
                "  'connector.username' = 'root', \n" +
                "  'connector.password' = '123456',\n" +
                "  'connector.write.flush.max-rows' = '1'" +
                ")");


        env.executeSql("insert into ads_mysql_sent_index" +
                " select prediction as sent , d ,count(*) as c from temp group by d , prediction");


        StreamOperator.execute();

    }
}


